package org.ngbx.demo.flink.ms.job;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.ngbx.demo.flink.ms.SimulationEvent;

import java.time.Duration;
import java.util.Iterator;
import java.util.Properties;

public class MessageSequentializationEventJob {
    // private static final Logger log = LoggerFactory.getLogger(MessageSequentializationEventJob.class);

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建kafka配置属性
        Properties kafkaProps = createKafkaProperties();

        KafkaSource<SimulationEvent> source = KafkaSource.<SimulationEvent>builder()
                .setProperties(kafkaProps)
                .setTopics("message_sequentialization_event")
                .setGroupId("message_sequentialization")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimulationEventDeserializationSchema())
                .build();

        WatermarkStrategy<SimulationEvent> watermarkStrategy = WatermarkStrategy.<SimulationEvent>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
                .withTimestampAssigner((SerializableTimestampAssigner<SimulationEvent>) (element, recordTimestamp) -> element.getOccurrenceTime());

        DataStream<SimulationEvent> simulationEventDS = env.fromSource(source, watermarkStrategy, "Message Sequentialization Source");

        // 周期性 发射watermark
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarks = simulationEventDS
                .map((MapFunction<SimulationEvent, Tuple2<String, Long>>) event -> {
                            System.out.println("接收到信息：" + event.toString());
                            return Tuple2.of(event.getKey(), event.getOccurrenceTime());
                        }
                )
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .assignTimestampsAndWatermarks(
                        new WatermarkStrategy<Tuple2<String, Long>>() {
                            private static final long serialVersionUID = -8873639694196414860L;

                            @Override
                            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(
                                    WatermarkGeneratorSupplier.Context context) {
                                return new WatermarkGenerator<Tuple2<String, Long>>() {
                                    private long maxTimeStamp = 0L;

                                    @Override
                                    public void onEvent(Tuple2<String, Long> event, long eventTimestamp,
                                                        WatermarkOutput output) {
                                        // 每次来一条数据就会触发一次
                                        maxTimeStamp = Math.max(maxTimeStamp, event.f1);
                                    }

                                    @Override
                                    public void onPeriodicEmit(WatermarkOutput output) {
                                        // 周期性 发射watermark
                                        // 允许的最大延迟时间
                                        long maxOutOfOrderNess = 5000L;
                                        output.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderNess));
                                    }
                                };
                            }
                        }.withTimestampAssigner(((element, recordTimestamp) -> element.f1))
                );

        watermarks.keyBy(x -> x.f0).window(TumblingEventTimeWindows.of(Time.seconds(6)))
                .apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {

                    private static final long serialVersionUID = 65693184846116387L;

                    @Override
                    public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input,
                                      Collector<String> out) {
                        Iterator<Tuple2<String, Long>> iterator = input.iterator();
                        int count = 0;
                        while (iterator.hasNext()) {
                            count++;
                            iterator.next();
                        }

                        System.out.println("================>>>>>>>>>>");

                        out.collect(window.getStart() + "->" + window.getEnd() + " " + s + ":" + count);
                    }
                }).print();

        env.execute();
    }

    private static Properties createKafkaProperties() {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
        // kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserClickedEventDeserializationSchema.class.getCanonicalName());
        return kafkaProps;
    }
}
